package com.lyloou.practicerabbitmq.config;

import com.lyloou.practicerabbitmq.entity.Const;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.HashMap;
import java.util.Map;

/**
 * @author lilou
 * @since 2021/2/18
 */
@Configuration
@EnableRabbit
public class RabbitConfig {

    @Bean
    public DirectMessageListenerContainer listenerContainer(ConnectionFactory factory) {
        DirectMessageListenerContainer container = new DirectMessageListenerContainer(factory);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setPrefetchCount(5);
        container.setConsumersPerQueue(5);
        container.setMessagesPerAck(1);
        container.setTaskExecutor(taskScheduler());
        return container;
    }

    @Bean
    public ThreadPoolTaskExecutor taskScheduler() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        return executor;
    }


    // 由json序列化对象
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    // rabbit 管理类
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory factory) {
        return new RabbitAdmin(factory);
    }

    // 订单队列
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable(Const.q.order).build();
    }

    // 延迟队列
    @Bean
    public Queue ttlQueue() {
        Map<String, Object> args = new HashMap<>();
        // 10000 毫秒到期后，添加到死信队列中
        args.put("x-message-ttl", 10000);
        args.put("x-dead-letter-exchange", Const.x.dlx);
        args.put("x-dead-letter-routing-key", Const.k.dlx);
        return new Queue(Const.q.ttl, true, false, false, args);
    }

    // 死信队列，取消订单
    @Bean
    public Queue dlxQueue() {
        Map<String, Object> args = new HashMap<>();
        return new Queue(Const.q.dlx, true, false, false, args);
    }


    // 订单队列交换器
    @Bean
    public Exchange orderExchange() {
        return new DirectExchange(Const.x.order, true, false, new HashMap<>());
    }

    // 延迟队列交换器
    @Bean
    public Exchange ttlExchange() {
        return new DirectExchange(Const.x.ttl, true, false, new HashMap<>());
    }

    // 死信队列交换器
    @Bean
    public Exchange dlxExchange() {
        return new DirectExchange(Const.x.dlx, true, false, new HashMap<>());
    }

    // 订单队列绑定器
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(Const.k.order).noargs();
    }

    // 延迟队列绑定器
    @Bean
    public Binding ttlBinding() {
        return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with(Const.k.ttl).noargs();
    }

    // 死信队列绑定器
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with(Const.k.dlx).noargs();
    }
}
